package cn.doitedu.demo4.utils;

import cn.doitedu.demo4.calculators.RuleCalculator;
import cn.doitedu.demo4.beans.RuleMetaBean;
import groovy.lang.GroovyClassLoader;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RuleCalculatorBuilder {

    public static RuleCalculator buildCalculator(RuleMetaBean ruleMetaBean, GroovyClassLoader groovyClassLoader) throws InstantiationException, IllegalAccessException {
        Class aClass = groovyClassLoader.parseClass(ruleMetaBean.getCalculator_source_code());
        RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();


        return ruleCalculator;
    }



    public static void restoreAllCalculator(ReadOnlyBroadcastState<String, RuleMetaBean> broadcastState,
                                            GroovyClassLoader groovyClassLoader,
                                            RuntimeContext runtimeContext,
                                            ConcurrentHashMap<String,RuleCalculator> calculatorPool) throws Exception {
        // 要重建故障重启前所有的运算机对象
        for (Map.Entry<String, RuleMetaBean> entry : broadcastState.immutableEntries()) {
            String ruleId = entry.getKey();
            RuleMetaBean ruleMetaBean = entry.getValue();

            // 根据元数据中的运算机代码，编译、加载，并反射实例化
            RuleCalculator ruleCalculator = RuleCalculatorBuilder.buildCalculator(ruleMetaBean, groovyClassLoader);

            // 对运算机初始化
            ruleCalculator.init(runtimeContext, ruleMetaBean.getRule_param(),ruleMetaBean.getPreSelectUserBitmap(),ruleMetaBean.getHistory_end_time());

            // 将初始化完成的规则运算机对象，放入运算机池
            calculatorPool.put(ruleId,ruleCalculator);
        }
    }



}
